package ds_industry_2025.ds.ds_03.sjwj

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{SparkSession, functions}

/*
    2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列名为商品id，
    按照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下
    。
 */
object t2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t2")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //  todo 读取第一题的结果
    val source = spark.table("dwd.t1")
      .withColumn("user_id",col("user_id").cast(DoubleType))

    //  todo 更改字段的值
    val data = source
      .withColumn(
        "sku_id",
        concat(lit("sku_id"), col("sku_id"))
      )


    //  todo 提取sku列
    val skus = data.select("sku_id").distinct()
      .orderBy(
        split(col("sku_id"), "id")(1).cast("int")
      )
      .map(_(0).toString)
      .collect()

    val result = data.groupBy("user_id")
      .pivot(col("sku_id"), skus)
      .agg(lit(1.0))
      .na.fill(0.0)
      .orderBy("user_id")

    result.show

    println("-------------第一行前5列结果展示为-----------------")
    result.limit(1)
      .select(result.columns.slice(0,5).map(col): _*)
      .foreach(
        r => println(r.mkString(","))
      )

    result.write.format("hive").saveAsTable("dwd.t2")


    spark.close()

  }

}
